弹力设计篇之"异步通讯设计" - [2026重制版]

{
  "specversion": "1.0",
  "type": "com.example.order.created",
  "source": "/orders-service",
  "id": "A234-1234-1234",
  "time": "2026-06-06T12:00:00Z",
  "datacontenttype": "application/json",
  "subject": "order-12345",
  "data": {
    "orderId": "ORD-20260606-001",
    "userId": "USER-789",
    "items": [
      {"productId": "PROD-001", "quantity": 2, "price": 99.99}
    ],
    "totalAmount": 199.98,
    "currency": "CNY",
    "status": "CREATED"
  },
  "extensions": {
    "traceparent": "00-0af7651916cd43dd8448eb211c80319c-b7ad6b716920581d-01",
    "partitionkey": "USER-789"
  }
}

# application.yml
spring:
  kafka:
    bootstrap-servers: kafka-0:9092,kafka-1:9092,kafka-2:9092
    producer:
      # 必须的可靠性配置
      acks: all                    # 等待所有ISR副本确认
      retries: 3                   # 重试次数
      retry-backoff-ms: 100        # 重试间隔
      max-in-flight-requests-per-connection: 5  # 控制重试顺序性
      
      # 性能优化
      batch-size: 16384            # 批次大小（16KB）
      linger-ms: 5                 # 等待更多消息组成批次
      buffer-memory: 33554432      # 缓冲区大小（32MB）
      compression-type: lz4       # 压缩算法
      
      # 序列化
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
      
      # Schema Registry
      schema.registry.url: http://schema-registry:8081
      auto-register-schemas: true
      
    consumer:
      group-id: order-service-group
      auto-offset-reset: earliest     # 最早未消费的消息
      enable-auto-commit: false       # 手动提交offset（确保精确一次）
      max-poll-records: 500           # 每次最大拉取数量
      session-timeout-ms: 30000       # 会话超时
      heartbeat-interval-ms: 10000    # 心跳间隔
      isolation-level: read_committed # 读取已提交的事务消息
      
      # 反序列化
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
      schema.registry.url: http://schema-registry:8081
      
      # 并发控制
      concurrency: 4                  # 并发消费者数量
      
    listener:
      ack-mode: MANUAL_IMMEDIATE      # 手动立即确认
      missing-topics-fatal: false     # topic不存在时不报错

@Service
@RequiredArgsConstructor
@Slf4j
public class OrderEventPublisher {

    private final KafkaTemplate<String, OrderCreatedEvent> kafkaTemplate;
    private final ApplicationEventPublisher applicationEventPublisher;

    /**
     * 发布订单创建事件
     * 使用事务确保本地操作和消息发布的原子性
     */
    @Transactional
    public void publishOrderCreated(Order order) {
        // 1. 构建CloudEvents标准事件
        OrderCreatedEvent event = OrderCreatedEvent.builder()
            .specVersion("1.0")
            .type("com.ecommerce.order.created")
            .source("/order-service")
            .id(UUID.randomUUID().toString())
            .time(Instant.now())
            .subject(order.getId())
            .data(OrderEventData.from(order))
            .extension("traceId", MDC.get("traceId"))
            .build();

        // 2. 发送到Kafka（使用事务）
        try {
            kafkaTemplate.executeInTransaction(operations -> {
                operations.send(
                    "order-events",                          // topic名称
                    order.getUserId(),                        // partition key（按用户分区）
                    event                                     // 事件对象
                ).addCallback(
                    result -> log.info("Event sent: offset={}", 
                                      result.getRecordMetadata().offset()),
                    failure -> log.error("Failed to send event", failure)
                );
                return true;
            });
            
            // 3. 发布本地Spring事件（用于同步处理或缓存更新）
            applicationEventPublisher.publishEvent(
                new OrderCreatedLocalEvent(order.getId())
            );
            
        } catch (Exception e) {
            log.error("Failed to publish order created event for order: {}", 
                      order.getId(), e);
            throw new EventPublishException("Failed to publish event", e);
        }
    }
}

// Avro Schema定义（OrderCreatedEvent.avsc）
/*
{
  "type": "record",
  "name": "OrderCreatedEvent",
  "namespace": "com.ecommerce.event.order",
  "fields": [
    {"name": "orderId", "type": "string"},
    {"name": "userId", "type": "string"},
    {"name": "items", "type": {
      "type": "array",
      "items": {
        "type": "record",
        "name": "OrderItem",
        "fields": [
          {"name": "productId", "type": "string"},
          {"name": "quantity", "type": "int"},
          {"name": "unitPrice", "type": "double"}
        ]
      }
    }},
    {"name": "totalAmount", "type": "double"},
    {"name": "currency", "type": "string"},
    {"name": "createdAt", "type": "long", "logicalType": "timestamp-millis"}
  ]
}
*/

@Service
@Slf4j
@RequiredArgsConstructor
public class OrderEventConsumer {

    private final InventoryService inventoryService;
    private final PaymentService paymentService;
    private final NotificationService notificationService;
    private final OrderEventRepository orderEventRepository;

    /**
     * 消费订单创建事件
     * 使用幂等性设计防止重复消费
     */
    @KafkaListener(
        topics = "order-events",
        groupId = "inventory-service-group",
        containerFactory = "kafkaListenerContainerFactory"
    )
    public void handleOrderCreated(
            @Payload OrderCreatedEvent event,
            @Headers Map<String, Object> headers,
            Acknowledgment acknowledgment) {

        String eventId = event.getId();
        
        try {
            // 1. 幂等性检查：避免重复处理
            if (orderEventRepository.existsByEventId(eventId)) {
                log.info("Event already processed, skipping: {}", eventId);
                acknowledgment.acknowledge();
                return;
            }

            // 2. 记录事件接收（用于追踪和对账）
            orderEventRepository.save(OrderEventRecord.received(event));

            // 3. 业务处理：扣减库存
            inventoryService.reserveStock(
                event.getData().getOrderId(),
                event.getData().getItems()
            );

            // 4. 标记事件已处理
            orderEventRepository.markAsProcessed(eventId);

            // 5. 手动确认消息
            acknowledgment.acknowledge();
            
            log.info("Successfully processed order created event: {}", eventId);
            
        } catch (InsufficientStockException e) {
            // 库存不足：记录失败但不确认消息（触发重试）
            log.error("Insufficient stock for order: {}", 
                     event.getData().getOrderId(), e);
            throw new RetryableException(e);  // 抛出异常触发重试
            
        } catch (Exception e) {
            // 其他错误：记录到死信队列
            log.error("Failed to process event: {}", eventId, e);
            sendToDeadLetterQueue(event, headers, e);
            acknowledgment.acknowledge();  // 确认消息，避免无限重试
        }
    }

    /**
     * 发送到死信队列（DLQ）
     */
    private void sendToDeadLetterQueue(
            OrderCreatedEvent event, 
            Map<String, Object> headers,
            Exception error) {
        
        DeadLetterEntry deadLetter = DeadLetterEntry.builder()
            .originalTopic("order-events")
            .event(event)
            .headers(headers)
            .errorClass(error.getClass().getName())
            .errorMessage(error.getMessage())
            .stackTrace(ExceptionUtils.getStackTrace(error))
            .failedAt(Instant.now())
            .retryCount(0)
            .build();
            
        deadLetterRepository.save(deadLetter);
        log.warn("Sent to dead letter queue: {}", event.getId());
    }
}

# kafka-cluster.yaml (使用Strimzi Operator)
apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
  namespace: messaging
spec:
  kafka:
    version: 3.9.0
    replicas: 3
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: false
      - name: tls
        port: 9093
        type: internal
        tls: true
    config:
      offsets.topic.replication.factor: 3
      transaction.state.log.replication.factor: 3
      transaction.state.log.min.isr: 2
      default.replication.factor: 3
      min.insync.replicas: 2
      log.retention.hours: 168          # 保留7天
      log.segment.bytes: 1073741824     # 1GB per segment
      log.retention.check.interval.ms: 300000  # 5分钟检查一次
      num.partitions: 10               # 默认分区数
      auto.create.topics.enable: false  # 禁止自动创建topic（安全考虑）
      message.max.bytes: 10485760       # 单条消息最大10MB
    storage:
      type: jbod
      volumes:
      - id: 0
        type: persistent-claim
        size: 100Gi
        class: standard              # StorageClass
        deleteClaim: false
    resources:
      requests:
        cpu: "2"
        memory: 4Gi
      limits:
        cpu: "4"
        memory: 8Gi
    jvmOptions:
      -Xms: 2g
      -Xmx: 4g
    rack:
      topologyKey: topology.kubernetes.io/zone  # 跨可用区分布
    # KRaft模式：Kafka 3.3+生产可用，Kafka 4.0（2025-03发布）起完全移除ZooKeeper
  entityOperator:
    topicOperator: {}
    userOperator: {}
---
# 创建订单事件topic（带多分区和副本）
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaTopic
metadata:
  name: order-events
  namespace: messaging
  labels:
    strimzi.io/cluster: my-cluster
spec:
  partitions: 12                    # 12个分区（支持并行消费）
  replication-factor: 3             # 3副本（高可用）
  config:
    retention.ms: 604800000         # 保留7天
    segment.bytes: 1073741824       # 1GB per segment
    cleanup.policy: delete          # 过期删除
    compression.type: producer     # 使用生产者指定的压缩算法
    min.insync.replicas: 2          # 至少2个副本写入成功才确认
---
# Schema Registry配置（Confluent或Apicurio）
apiVersion: apps/v1
kind: Deployment
metadata:
  name: schema-registry
  namespace: messaging
spec:
  replicas: 3
  selector:
    matchLabels:
      app: schema-registry
  template:
    metadata:
      labels:
        app: schema-registry
    spec:
      containers:
      - name: schema-registry
        image: confluentinc/cp-schema-registry:7.5.0
        ports:
        - containerPort: 8081
        env:
        - name: SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS
          value: "my-cluster-kafka-bootstrap:9092"
        - name: SCHEMA_REGISTRY_HOST_NAME
          value: schema-registry
        - name: SCHEMA_REGISTRY_LOG_LEVEL
          value: INFO
        resources:
          requests:
            cpu: 500m
            memory: 1Gi
          limits:
            cpu: "1"
            memory: 2Gi
---
apiVersion: v1
kind: Service
metadata:
  name: schema-registry
  namespace: messaging
spec:
  selector:
    app: schema-registry
  ports:
  - port: 8081
    targetPort: 8081
  type: ClusterIP

@Service
@Slf4j
@RequiredArgsConstructor
public class OrderSagaOrchestrator {

    private final KafkaTemplate<String, SagaEvent> kafkaTemplate;
    private final SagaStateRepository sagaStateRepository;

    /**
     * 启动订单Saga流程
     */
    public void startOrderSaga(Order order) {
        String sagaId = UUID.randomUUID().toString();
        
        // 初始化Saga状态
        sagaStateRepository.save(SagaState.builder()
            .sagaId(sagaId)
            .order(order)
            .currentStep(SagaStep.ORDER_CREATED)
            .status(SagaStatus.STARTED)
            .startedAt(Instant.now())
            .build());

        // 发布第一个步骤事件
        publishSagaEvent(sagaId, SagaEventType.RESERVE_INVENTORY, order);
    }

    /**
     * 处理Saga步骤完成事件
     */
    @KafkaListener(topics = "saga-reply-events", groupId = "saga-orchestrator")
    public void handleStepCompleted(SagaReplyEvent reply) {
        SagaState state = sagaStateRepository.findBySagaId(reply.getSagaId())
            .orElseThrow(() -> new SagaNotFoundException(reply.getSagaId()));

        switch (reply.getStep()) {
            case RESERVE_INVENTORY:
                if (reply.isSuccess()) {
                    state.setCurrentStep(SagaStep.INVENTORY_RESERVED);
                    publishSagaEvent(state.getSagaId(), SagaEventType.PROCESS_PAYMENT, state.getOrder());
                } else {
                    // 库存不足，取消订单
                    state.setStatus(SagaStatus.COMPENSATING);
                    publishSagaEvent(state.getSagaId(), SagaEventType.CANCEL_ORDER, state.getOrder());
                }
                break;

            case PROCESS_PAYMENT:
                if (reply.isSuccess()) {
                    state.setCurrentStep(SagaStep.PAYMENT_PROCESSED);
                    publishSagaEvent(state.getSagaId(), SagaEventType.CONFIRM_ORDER, state.getOrder());
                } else {
                    // 支付失败，补偿库存
                    state.setStatus(SagaStatus.COMPENSATING);
                    publishCompensationEvent(state.getSagaId(), CompensationType.RELEASE_INVENTORY, state.getOrder());
                }
                break;

            case CONFIRM_ORDER:
                state.setStatus(SagaStatus.COMPLETED);
                state.setCompletedAt(Instant.now());
                log.info("Saga completed successfully: {}", state.getSagaId());
                break;

            case COMPENSATION_COMPLETED:
                state.setStatus(SagaStatus.COMPENSATED);
                log.info("Saga compensated successfully: {}", state.getSagaId());
                break;
        }

        sagaStateRepository.save(state);
    }

    private void publishSagaEvent(String sagaId, SagaEventType eventType, Order order) {
        SagaEvent event = SagaEvent.builder()
            .sagaId(sagaId)
            .eventType(eventType)
            .payload(order)
            .timestamp(Instant.now())
            .build();
            
        kafkaTemplate.send("saga-command-events", sagaId, event);
    }
}

用户下单 → 创建订单(50ms) → 扣减库存(30ms) → 支付(1000ms) → 发送短信(200ms) → 积分奖励(20ms) → 返回结果
总耗时: ~1300ms

# knative-eventing-source.yaml
# 使用Knative Eventing构建Serverless事件驱动架构
apiVersion: sources.knative.dev/v1
kind: ApiServerSource
metadata:
  name: order-events-source
  namespace: production
spec:
  serviceAccountName: events-sa
  mode: Resource
  resources:
  - apiVersion: v1
    kind: Event
  sink:
    ref:
      apiVersion: eventing.knative.dev/v1
      kind: Broker
      name: default
---
apiVersion: eventing.knative.dev/v1
kind: Trigger
metadata:
  name: inventory-trigger
  namespace: production
spec:
  broker: default
  filter:
    attributes:
      type: com.ecommerce.order.created
  subscriber:
    ref:
      apiVersion: serving.knative.dev/v1
      kind: Service
      name: inventory-handler
---
apiVersion: serving.knative.dev/v1
kind: Service
metadata:
  name: inventory-handler
  namespace: production
spec:
  template:
    metadata:
      annotations:
        autoscaling.knative.dev/minScale: "0"   # 允许缩容到0
        autoscaling.knative.dev/maxScale: "100" # 最大100个实例
        autoscaling.knative.dev/target: "10"   # 目标并发数
    spec:
      containers:
      - image: inventory-handler:v2.0
        env:
        - name: KAFKA_BROKERS
          value: "my-cluster-kafka-bootstrap:9092"
        resources:
          requests:
            cpu: 100m
            memory: 128Mi

-- 实时计算每个商品的销售额（持续查询）
CREATE MATERIALIZED VIEW product_sales AS
SELECT
    window_start(product_created_at, INTERVAL '1 MINUTE') AS minute,
    product_id,
    COUNT(*) AS order_count,
    SUM(total_amount) AS total_revenue,
    AVG(total_amount) AS avg_order_value
FROM order_events  -- 直接订阅Kafka topic
WHERE event_type = 'ORDER_CREATED'
GROUP BY
    window_start(product_created_at, INTERVAL '1 MINUTE'),
    product_id;

-- 结果可以像普通表一样查询
SELECT * FROM product_sales WHERE minute > NOW() - INTERVAL '5 MINUTES';

// Rust编写的WASM消息处理函数
use wasm_bindgen::prelude::*;
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize)]
struct OrderEvent {
    order_id: String,
    user_id: String,
    total_amount: f64,
}

#[wasm_bindgen]
pub fn process_order_event(event_json: &str) -> String {
    // 解析输入事件
    let event: OrderEvent = serde_json::from_str(event_json).unwrap();
    
    // 业务逻辑：判断是否为VIP用户的大额订单
    let is_priority = event.total_amount > 1000.0;
    
    // 输出路由决策
    let routing_decision = serde_json::json!({
        "target_queue": if is_priority { "priority-processing" } else { "normal-processing" },
        "priority": if is_priority { "high" } else { "normal" },
        "requires_approval": is_priority && event.total_amount > 5000.0
    });
    
    routing_decision.to_string()
}